package com.tuanzhang.zhw.rpc.server;

import com.tuanzhang.zhw.rpc.codec.RpcRequest;
import com.tuanzhang.zhw.rpc.codec.RpcResponse;
import io.netty.channel.*;
import io.netty.util.concurrent.EventExecutorGroup;
import lombok.extern.slf4j.Slf4j;
import net.sf.cglib.reflect.FastClass;
import net.sf.cglib.reflect.FastMethod;

import java.lang.reflect.InvocationTargetException;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
public class RpcServerHandler extends SimpleChannelInboundHandler<RpcRequest> {

    /**
     * interfaceName-interfaceImplClass Bean
     */
    private Map<String, Object> handlerMap;

    /**
     * 任务线程池: 用于异步提交任务, 从而不阻塞worker线程
     */
    private ThreadPoolExecutor executor = new ThreadPoolExecutor(16, 16, 600, TimeUnit.SECONDS, new ArrayBlockingQueue<>(65536));


    public RpcServerHandler(Map<String, Object> handlerMap) {
        this.handlerMap = handlerMap;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest rpcRequest) throws Exception {
        executor.submit(() -> {
            RpcResponse rpcResponse = new RpcResponse();
            rpcResponse.setRequestId(rpcRequest.getRequestId());
            try {
                Object result = handler(rpcRequest);
                rpcResponse.setResult(result);
            } catch (Throwable e) {
                rpcResponse.setThrowable(e);
                log.error("rpc 调用错误", e);
            }

            // 添加handle后置处理逻辑
            ctx.writeAndFlush(rpcResponse).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if(future.isSuccess()) {
                        // afterRpcHook
                    }
                }
            });

        });
    }


    /**
     * 具体调用方法
     *
     * @param request
     * @return
     * @throws InvocationTargetException
     */
    private Object handler(RpcRequest request) throws InvocationTargetException {
        String className = request.getClassName();
        String methodName = request.getMethodName();
        Class<?>[] parameterTypes = request.getParameterTypes();
        Object[] parameters = request.getParameters();

        // handlerMap中找到具体接口名称
        Object serviceRef = handlerMap.get(className);
        Class<?> serviceRefClass = serviceRef.getClass();

        // cglib 反射
        FastClass fastClass = FastClass.create(serviceRefClass);
        FastMethod method = fastClass.getMethod(methodName, parameterTypes);
        Object invokeResult = method.invoke(serviceRef, parameters);
        return invokeResult;
    }

    /**
     * 异常处理：关闭连接
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("rpc 服务端异常", cause);
        ctx.close();
    }
}
